package com.lc.cms.kafka;

import com.alibaba.fastjson.JSONObject;
import com.lc.cms.redis.RedisUtil;
import com.lc.cms.utils.SystemConstant;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@Log4j2
public class ConsumerListener {

    private final RedisUtil redisUtil;

    @Autowired
    public ConsumerListener(RedisUtil redisUtil) {
        this.redisUtil = redisUtil;
    }

    @KafkaListener(topics = {SystemConstant.TOPIC_STATION_PROTOCOL}, groupId = "official")
    public void listen(ConsumerRecord<String, String> record) {
        try {
            log.info("topic = {}, offset = {}, value = {}", record.topic(), record.offset(), record.value());
            redisUtil.set("protocol", record.value());
            String protocol = record.value();
            log.info(protocol);
        } catch (Exception e) {
            log.error("topic = {}, offset = {}, value = {}", record.topic(), record.offset(), record.value());
        }
    }
}
